package com.jing.cloud.service.chat.websocket;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.jing.cloud.service.chat.adapter.MessageAdapter;
import com.jing.cloud.service.chat.adapter.MessageBean;
import com.jing.cloud.service.chat.adapter.helper.MessageHelper;
import com.jing.cloud.service.chat.bean.Message;
import com.jing.cloud.service.chat.bean.UserMessage;
import com.jing.cloud.service.chat.constant.ChatConstant;
import com.jing.cloud.service.chat.model.SessionCache;
import com.jing.cloud.service.util.bean.StringUtil;
import com.jing.cloud.service.util.keygen.DefaultKeyGenerator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.util.List;
import java.util.Map;

/**
 * Created by 29017 on 2017/9/8.
 */
@Component
@Slf4j
public class ChatHandler extends TextWebSocketHandler {
    private static final Map<String, WebSocketSession> sessionMap = Maps.newConcurrentMap();

    //    private static final Executor executor = Executors.newFixedThreadPool(200);
    @Autowired
    private MessageHelper messageHelper;

    @Override
    public void afterConnectionEstablished(final WebSocketSession session) throws Exception {
        SessionCache cache = getCache(session);
        long userMessageId = 0;
        while (true) {
            if (!session.isOpen()) {
                return;
            }
            List<UserMessage> ums = messageHelper.list(cache.getUserId(), userMessageId, cache.getDeviceType());
            if (ums.isEmpty()) {
                break;
            }
            for (UserMessage um : ums) {
                Message msg = um.getMessage();
                if (null == msg) {
                    break;
                }
                MessageBean messageBean = MessageBean.newInstance(msg);
                pushMessage(session, messageBean);
                userMessageId = um.getId();
            }
        }
        put(session);
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String str = message.getPayload();
        log.info(str);
        try {
            MessageBean messageBean = JSON.parseObject(str, MessageBean.class);
            messageBean.setId(DefaultKeyGenerator.getInstance().generateKey().longValue());
            SessionCache sessionCache = getCache(session);
            messageBean.setSender(sessionCache.getUserId());
            messageBean.setTicket(sessionCache.getTicket());
            messageBean.setDeviceType(sessionCache.getDeviceType());
            MessageAdapter.handleMessage(messageBean);
        } catch (Exception e) {
            log.error("handleMessage error ", e);
        }
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        remove(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        remove(session);
    }

    private static WebSocketSession get(String ticket) {
        return sessionMap.get(ticket);
    }

    private static void put(WebSocketSession session) {
        SessionCache cache = getCache(session);
        String ticket = cache.getTicket();
        remove(ticket);
        sessionMap.put(ticket, session);
        log.info("{} connected", cache);
    }

    private static void remove(WebSocketSession session) {
        String ticket = getCache(session).getTicket();
        remove(ticket);
    }

    private static void remove(String ticket) {
        WebSocketSession oldSession = sessionMap.get(ticket);
        if (null != oldSession && oldSession.isOpen()) {
            try {
                oldSession.close(CloseStatus.GOING_AWAY.withReason("kick out."));
            } catch (Exception e) {
                log.error("{} remove exception", ticket, e);
            }
            log.info("{} removed", getCache(oldSession));
        }
        sessionMap.remove(ticket);
    }


    private static SessionCache getCache(WebSocketSession session) {
        return (SessionCache) session.getAttributes().get(ChatConstant.SESSION_CACHE_ATTR_STR);
    }

    public static void pushMsg(String targetTicket, MessageBean messageBean) {

        WebSocketSession session = get(targetTicket);
        if (null == session || !session.isOpen()) {
            log.info("ticket[{}] 连接已断开。不推送", targetTicket);
            return;
        }
        pushMessage(session, messageBean);

    }

    private static void pushMessage(WebSocketSession session, MessageBean messageBean) {
        try {
            SessionCache c = getCache(session);
            String targetTicket = c.getTicket();
            if (null == session || !session.isOpen()) {
                log.info("ticket[{}] 连接已断开。不推送", targetTicket);
                return;
            }
            MessageBean m2 = new MessageBean();
            BeanUtils.copyProperties(messageBean, m2);
            m2.setTicket(null);//ticket 不传递给app
            String json = StringUtil.toJson(m2);
            TextMessage textMessage = new TextMessage(json);
            session.sendMessage(textMessage);
            log.info("push message to user[ticket{}],messageId[{}]", targetTicket, messageBean.getId());
        } catch (Exception e) {
            log.error("send2User error", e);
        }
    }
}
